package com.hh.game.service.gateway.pool;

import cn.hutool.core.util.StrUtil;
import com.hh.game.common.base.exception.GameException;
import com.hh.game.common.base.manager.session.Session;
import com.hh.game.service.gateway.codec.websocket.WebSocketRequest;
import com.hh.game.service.gateway.codec.websocket.WebSocketResponse;
import com.hh.game.service.gateway.manager.RouteEnum;
import com.hh.game.service.gateway.manager.RouteInfo;
import com.hh.game.service.gateway.manager.RpcManager;
import com.hh.game.service.gateway.codec.websocket.InputMessage;
import com.hh.game.service.gateway.codec.websocket.OutputMessage;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class GatewayWorkPool {
    private final static Logger LOGGER = LoggerFactory.getLogger(GatewayWorkPool.class);

    private final static String SESSION_INPUT_MESSAGE_KEY = "SESSION_INPUT_MESSAGE_KEY";

    private final static ThreadLocal<WebSocketRequest> THREAD_LOCAL = new ThreadLocal();

    private final static ThreadGroup threadGroup = new ThreadGroup("gateway-work-thread-group");
    private final static AtomicInteger threadNumber = new AtomicInteger(1);
    private static ExecutorService executorService;

    public static void shutdown() {
        executorService.shutdown();
    }

    public static void addWork(WebSocketRequest request) {
        THREAD_LOCAL.set(request);
        executorService.submit(() -> {
            OutputMessage msg = null;
            GameException exception = null;
            InputMessage in = request.getInputMessage();
            String route = in.getRoute();
            try {
                String[] args = route.split("\\.");
                RouteInfo routeInfo = new RouteInfo.RouteInfoBuilder()
                        .serviceName(args[0])
                        .handler(args[1])
                        .group(null)
                        .method(args[2])
                        .version(in.getVersion())
                        .bytes(in.getBytes())
                        .data(in.getData())
                        .stringData(in.getStringData())
                        .isByte(in.isByte())
                        .isString(in.isString())
                        .build();
                Object result = RpcManager.invoke(routeInfo);
                msg = new OutputMessage(route, result);
            } catch (GameException gameException) {
                LOGGER.error(gameException.getMessage());
                exception = gameException;
            } catch (RuntimeException rex) {
                String throwableMsg = rex.getMessage();
                GameException gameException = GameException.parse(subCRLF(throwableMsg));
                if (gameException != null) {
                    LOGGER.error(gameException.getMessage());
                    gameException.setRoute(in.getRoute());
                    exception = gameException;
                } else {
                    LOGGER.error(rex.getMessage());
                    exception = buildGameException(in.getRoute());
                }
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
                exception = buildGameException(in.getRoute());
            }
            if (exception != null) {
                msg = new OutputMessage(RouteEnum.EXCEPTION.route(), exception.getClientResult());
            }
            WebSocketResponse response = new WebSocketResponse(msg , request.session() ,request.getChannelHandlerContext());
            request.getChannelHandlerContext().channel().writeAndFlush(response);
        });
    }

    private static GameException buildGameException(String route) {
        GameException ex = new GameException(GameException.SERVER_ERROR, route, "server error ,try again later");
        return ex;
    }

    /**
     * 去掉换行后的字符串
     *
     * @param val
     * @return
     */
    private static String subCRLF(String val) {
        int index;
        if (val != null
                && ((index = val.indexOf("\r")) > -1 || (index = val.indexOf("\n")) > -1)) {
            return val.substring(0, index);
        }
        return val;
    }

    public static void init(int workThreadSize, int workQueueSize) {
        executorService = new ThreadPoolExecutor(workThreadSize, workThreadSize,
                0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(workQueueSize), // 使用有界队列，避免OOM

                //threadFactory
                r -> new Thread(threadGroup, r,
                        "gateway-work-thread-pool-" + threadNumber.getAndIncrement())
                ,
                //RejectedExecutionHandler
                (Runnable r, ThreadPoolExecutor executor) -> {
                    WebSocketRequest request = THREAD_LOCAL.get();
                    InputMessage inputMessage = request.getInputMessage();
                    LOGGER.info(StrUtil.format("gateway work queue full ,request info => [%]",
                            (inputMessage.isByte() ? inputMessage.getBytes() : inputMessage.getData())));
                    request.session().sendClient(RouteEnum.SERVER_WORK_POOL_FULL.route() ,"try again later");
                });
    }
}
